fix(convex): stop retrying a deleted table and explain it by Gilbert09 · Pull Request #91365 · PostHog/posthog

Gilbert09 · 2026-08-30T11:02:26Z

Problem

A Convex sync whose table was deleted on the source retried the same failing request on every schedule and showed the customer raw driver text they could not act on. The error never reached triage, because it is caught before it becomes an exception.

A sync only calls list_snapshot / document_deltas. Convex answers those with a 404 when a table that schema discovery listed no longer exists at read time: it was deleted on the source, or it is a component table that streaming export does not serve. The 404 was unclassified, so it retried and stored the bare 404 ... for url: <deployment host> string as latest_error.

Changes

A missing Convex table now fails fast with "PostHog couldn't find this table in your Convex deployment...", instead of retrying every run and showing raw driver text.
Classify 404 Client Error as non-retryable in the Convex source. Cloudflare surfaces transient edge problems as the 52x/530 family, which retry separately, so a 404 is never a transient blip this could disable a sync over.
Mechanical: two test cases assert the 404 is classified non-retryable and surfaces the actionable message.

How did you test this code?

Ran the Convex source tests (convex/tests/test_convex.py). Added a test_known_errors_match case for the missing-table 404 and test_missing_table_404_surfaces_actionable_message, which catches a revert of this bucket to a raw or None value. Did not run the full warehouse suite, because this sandbox has no dev stack.

Automatic notifications

Publish to changelog?

Docs update

None.

🤖 Agent context

Autonomy: Fully autonomous

Found while triaging a batch of production warehouse sync failure classes. Skills invoked: /writing-user-facing-copy, /writing-tests, /writing-code-comments, /writing-pr-descriptions.

The example named one table in an installed betterAuth component. The fix is intentionally not table- or component-specific, because the next step (turn off syncing for the table) is the same whether the table was deleted or is not served by streaming export. No customer values are in this diff; the test uses an invented deployment host.

Created with PostHog Desktop

A Convex sync only calls list_snapshot / document_deltas, which answer 404 when a table that schema discovery listed no longer exists at read time. The 404 was unclassified, so every scheduled run retried it and the customer saw the raw driver text (which carries the deployment host). Classify it non-retryable with an actionable message. Cloudflare transients are the 52x/530 family, so a 404 is never a transient blip this could disable a sync over. Generated-By: PostHog Desktop Task-Id: c515c2ae-d0fd-42c5-81fc-44b3650ff0ae

trunk-io · 2026-08-30T11:02:30Z

😎 Merged successfully - details.

github-actions · 2026-08-30T11:02:38Z

Hey @Gilbert09! 👋

It looks like your git author email on this PR isn't your @posthog.com address (owerstom@gmail.com). Since you're on the PostHog team, it's worth pointing your local git author email at your @posthog.com address. Why it matters:

Consistent work identity in git history — internal tooling that attributes commits to team members keys off your @posthog.com address.
Keeps team contributions easy to tell apart from external community ones when scanning history.

You can fix it for this repo with:

git config user.email "you@posthog.com"

Or set it globally with git config --global user.email "you@posthog.com". No need to redo this PR — just a nudge for next time. 🙂

github-actions · 2026-08-30T11:04:10Z

🤖 CI report

⚠️

Trunk lane — backend Python lane

This PR is assigned to the backend Python lane. It runs backend Python tests and may merge in parallel with PRs in other lanes.

ℹ️ ClickHouse migration SQL — 1 migration(s)

ClickHouse migration SQL per cloud environment

unset

all

CREATE TABLE IF NOT EXISTS posthog_test.logs34
(
    `time_bucket` DateTime MATERIALIZED toStartOfDay(timestamp),
    `original_expiry_timestamp` DateTime64(6),
    `uuid` String,
    `team_id` Int32,
    `trace_id` String,
    `span_id` String,
    `trace_flags` Int32,
    `timestamp` DateTime64(6) CODEC(DoubleDelta),
    `observed_timestamp` DateTime64(6),
    `created_at` DateTime64(6) MATERIALIZED now(),
    `body` String,
    `severity_text` LowCardinality(String),
    `severity_number` Int32,
    `service_name` LowCardinality(String),
    `resource_attributes` Map(LowCardinality(String), String),
    `resource_fingerprint` UInt64 MATERIALIZED cityHash64(resource_attributes),
    `instrumentation_scope` String,
    `event_name` String,
    `attributes_map_str` Map(LowCardinality(String), String),
    `level` String ALIAS severity_text,
    `mat_body_ipv4_matches` Array(String) ALIAS extractAll(body, '(\d\.((25[0-5]|(2[0-4]|1(0, 1)[0-9])(0, 1)[0-9])\.)(2, 2)([0-9]))'),
    `time_minute` DateTime ALIAS toStartOfMinute(timestamp),
    `attributes` Map(LowCardinality(String), String) ALIAS mapApply((k, v) -> (left(k, -5), v), attributes_map_str),
    `attributes_map_float` Map(LowCardinality(String), Float64) MATERIALIZED mapFilter((k, v) -> (v IS NOT NULL), mapApply((k, v) -> (concat(left(k, -5), '__float'), toFloat64OrNull(v)), attributes_map_str)),
    `attributes_map_datetime` Map(LowCardinality(String), DateTime64(6)) MATERIALIZED mapFilter((k, v) -> (v IS NOT NULL), mapApply((k, v) -> (concat(left(k, -5), '__datetime'), parseDateTimeBestEffortOrNull(v, 6)), attributes_map_str)),
    `_partition` UInt32,
    `_topic` String,
    `_offset` UInt64,
    `_bytes_uncompressed` UInt64,
    `_bytes_compressed` UInt64,
    `_record_count` UInt64,
    `pattern` String,
    `pattern_version` UInt8,
    INDEX idx_severity_text_set severity_text TYPE set(10) GRANULARITY 1,
    INDEX idx_attributes_str_keys mapKeys(attributes_map_str) TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_attributes_str_values mapValues(attributes_map_str) TYPE bloom_filter(0.001) GRANULARITY 1,
    INDEX idx_mat_body_ipv4_matches mat_body_ipv4_matches TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_body_ngram3 lower(body) TYPE ngrambf_v1(3, 25000, 2, 0) GRANULARITY 1,
    INDEX idx_uuid_bloom uuid TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_observed_minmax observed_timestamp TYPE minmax GRANULARITY 1,
    INDEX idx_timestamp_minmax timestamp TYPE minmax GRANULARITY 1,
    PROJECTION projection_aggregate_counts
    (
        SELECT
            team_id,
            time_bucket,
            toStartOfMinute(timestamp),
            service_name,
            severity_text,
            resource_fingerprint,
            count() AS event_count
        GROUP BY
            team_id,
            time_bucket,
            toStartOfMinute(timestamp),
            service_name,
            severity_text,
            resource_fingerprint
    )
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/noshard/posthog.logs34', '{replica}-{shard}')
PARTITION BY toDate(original_expiry_timestamp)
PRIMARY KEY (team_id, time_bucket, service_name, resource_fingerprint, severity_text, timestamp)
ORDER BY (team_id, time_bucket, service_name, resource_fingerprint, severity_text, timestamp)
TTL original_expiry_timestamp
SETTINGS
    allow_experimental_reverse_key = 1,
    index_granularity_bytes = 104857600,
    index_granularity = 8192,
    ttl_only_drop_parts = 1,
    add_minmax_index_for_numeric_columns = 1,
    map_serialization_version = 'with_buckets'

CREATE TABLE IF NOT EXISTS posthog_test.log_attributes2
(
    `team_id` Int32,
    `time_bucket` DateTime64(0),
    `service_name` LowCardinality(String),
    `resource_fingerprint` UInt64 DEFAULT 0,
    `attribute_key` LowCardinality(String),
    `attribute_value` String CODEC(ZSTD(5)),
    `attribute_count` SimpleAggregateFunction(sum, UInt64),
    `attribute_type` LowCardinality(String) DEFAULT 'log',
    `original_expiry_time_bucket` DateTime DEFAULT now(),
    INDEX idx_attribute_key attribute_key TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_attribute_value attribute_value TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_attribute_key_n3 attribute_key TYPE ngrambf_v1(3, 32768, 3, 0) GRANULARITY 1,
    INDEX idx_attribute_value_n3 attribute_value TYPE ngrambf_v1(3, 32768, 3, 0) GRANULARITY 1
)
ENGINE = ReplicatedAggregatingMergeTree('/clickhouse/tables/noshard/posthog.log_attributes2', '{replica}-{shard}')
PARTITION BY toDate(original_expiry_time_bucket)
ORDER BY (team_id, attribute_type, time_bucket, resource_fingerprint, attribute_key, attribute_value)
TTL original_expiry_time_bucket
SETTINGS
    deduplicate_merge_projection_mode = 'drop',
    index_granularity = 8192

CREATE TABLE IF NOT EXISTS posthog_test.logs_billing_metrics
(
    `team_id` Int32,
    `time_bucket` DateTime64(0),
    `service_name` LowCardinality(String),
    `bytes_uncompressed` SimpleAggregateFunction(sum, UInt64),
    `bytes_compressed` SimpleAggregateFunction(sum, UInt64),
    `record_count` SimpleAggregateFunction(sum, UInt64)
)
ENGINE = ReplicatedAggregatingMergeTree('/clickhouse/tables/noshard/posthog.logs_billing_metrics', '{replica}-{shard}')
PARTITION BY toYYYYMM(time_bucket)
ORDER BY (team_id, time_bucket, service_name)
SETTINGS
    deduplicate_merge_projection_mode = 'rebuild',
    index_granularity = 8192

CREATE TABLE IF NOT EXISTS posthog_test.logs_kafka_metrics
(
    `_partition` UInt32,
    `_topic` String,
    `max_offset` SimpleAggregateFunction(max, UInt64),
    `max_observed_timestamp` SimpleAggregateFunction(max, DateTime64(9)),
    `max_timestamp` SimpleAggregateFunction(max, DateTime64(9)),
    `max_created_at` SimpleAggregateFunction(max, DateTime64(9)),
    `max_lag` SimpleAggregateFunction(max, UInt64)
)
ENGINE = ReplicatedAggregatingMergeTree('/clickhouse/tables/noshard/posthog.logs_kafka_metrics', '{replica}-{shard}')
ORDER BY (_topic, _partition)

CREATE OR REPLACE TABLE posthog_test.logs_distributed AS posthog_test.logs34 ENGINE = Distributed('posthog_single_shard', 'posthog_test', 'logs34')

CREATE OR REPLACE TABLE posthog_test.log_attributes_distributed AS posthog_test.log_attributes2 ENGINE = Distributed('posthog_single_shard', 'posthog_test', 'log_attributes2')

CREATE TABLE IF NOT EXISTS posthog_test.logs_billing_metrics_distributed AS posthog_test.logs_billing_metrics ENGINE = Distributed('posthog_single_shard', 'posthog_test', 'logs_billing_metrics')

CREATE TABLE IF NOT EXISTS posthog_test.logs_kafka_metrics_distributed AS posthog_test.logs_kafka_metrics ENGINE = Distributed('posthog_single_shard', 'posthog_test', 'logs_kafka_metrics')

CREATE MATERIALIZED VIEW IF NOT EXISTS posthog_test.logs34_to_log_attributes TO posthog_test.log_attributes2
(
    `team_id` Int32,
    `time_bucket` DateTime64(0),
    `original_expiry_time_bucket` DateTime64(0),
    `service_name` LowCardinality(String),
    `resource_fingerprint` UInt64,
    `attribute_key` LowCardinality(String),
    `attribute_value` String,
    `attribute_type` LowCardinality(String),
    `attribute_count` SimpleAggregateFunction(sum, UInt64)
)
AS SELECT
    team_id,
    time_bucket,
    original_expiry_time_bucket,
    service_name,
    resource_fingerprint,
    attribute_key,
    attribute_value,
    attribute_type,
    attribute_count
FROM
(
    SELECT
        team_id AS team_id,
        toStartOfInterval(timestamp, toIntervalMinute(10)) AS time_bucket,
        toStartOfInterval(original_expiry_timestamp, toIntervalMinute(10)) AS original_expiry_time_bucket,
        service_name AS service_name,
        resource_fingerprint,
        mapFilter((k, v) -> ((length(k) < 256) AND (length(v) < 256)), attributes) AS attributes,
        arrayJoin(attributes) AS attribute,
        'log' AS attribute_type,
        attribute.1 AS attribute_key,
        attribute.2 AS attribute_value,
        sumSimpleState(1) AS attribute_count
    FROM posthog_test.logs34
    GROUP BY
        team_id,
        time_bucket,
        original_expiry_time_bucket,
        service_name,
        resource_fingerprint,
        attributes
)

CREATE MATERIALIZED VIEW IF NOT EXISTS posthog_test.logs34_to_resource_attributes TO posthog_test.log_attributes2
(
    `team_id` Int32,
    `time_bucket` DateTime64(0),
    `original_expiry_time_bucket` DateTime64(0),
    `service_name` LowCardinality(String),
    `resource_fingerprint` UInt64,
    `attribute_key` LowCardinality(String),
    `attribute_value` String,
    `attribute_type` LowCardinality(String),
    `attribute_count` SimpleAggregateFunction(sum, UInt64)
)
AS SELECT
    team_id,
    time_bucket,
    original_expiry_time_bucket,
    service_name,
    resource_fingerprint,
    attribute_key,
    attribute_value,
    attribute_type,
    attribute_count
FROM
(
    SELECT
        team_id AS team_id,
        toStartOfInterval(timestamp, toIntervalMinute(10)) AS time_bucket,
        toStartOfInterval(original_expiry_timestamp, toIntervalMinute(10)) AS original_expiry_time_bucket,
        service_name AS service_name,
        resource_fingerprint,
        arrayJoin(resource_attributes) AS attribute,
        'resource' AS attribute_type,
        attribute.1 AS attribute_key,
        attribute.2 AS attribute_value,
        sumSimpleState(1) AS attribute_count
    FROM posthog_test.logs34
    GROUP BY
        team_id,
        time_bucket,
        original_expiry_time_bucket,
        service_name,
        resource_fingerprint,
        resource_attributes
)

CREATE TABLE IF NOT EXISTS posthog_test.kafka_logs_avro
(
    `uuid` String,
    `trace_id` String,
    `span_id` String,
    `trace_flags` Int32,
    `timestamp` DateTime64(6),
    `observed_timestamp` DateTime64(6),
    `body` String,
    `severity_text` String,
    `severity_number` Int32,
    `service_name` String,
    `resource_attributes` Map(LowCardinality(String), String),
    `instrumentation_scope` String,
    `event_name` String,
    `attributes` Map(LowCardinality(String), String),
    `retention_days` Nullable(Int32),
    `pattern` Nullable(String),
    `pattern_version` Nullable(Int32)
)
ENGINE = Kafka(warpstream_logs, kafka_topic_list = 'clickhouse_logs', kafka_group_name = 'clickhouse-logs-avro-new', kafka_format = 'Avro')
SETTINGS
    kafka_skip_broken_messages = 100,
    kafka_num_consumers = 8,
    kafka_poll_timeout_ms = 3000,
    kafka_poll_max_batch_size = 1000,
    kafka_thread_per_consumer = 1,
    input_format_avro_allow_missing_fields = 1

CREATE MATERIALIZED VIEW IF NOT EXISTS posthog_test.kafka_logs34_avro_mv TO posthog_test.logs34
(
    `uuid` String,
    `trace_id` String,
    `span_id` String,
    `trace_flags` Int32,
    `timestamp` DateTime64(6),
    `observed_timestamp` DateTime64(6),
    `body` String,
    `severity_text` String,
    `severity_number` Int32,
    `service_name` String,
    `instrumentation_scope` String,
    `event_name` String,
    `attributes_map_str` Map(String, String),
    `resource_attributes` Map(String, String),
    `team_id` Int32,
    `original_expiry_timestamp` DateTime64(6),
    `_partition` UInt64,
    `_topic` LowCardinality(String),
    `_offset` UInt64,
    `_record_count` Int64,
    `_bytes_uncompressed` Nullable(Int64),
    `_bytes_compressed` Nullable(Int64),
    `pattern` String,
    `pattern_version` UInt8
)
AS SELECT
    uuid,
    trace_id,
    span_id,
    trace_flags,
    timestamp,
    observed_timestamp,
    body,
    severity_text,
    severity_number,
    service_name,
    instrumentation_scope,
    event_name,
    mapSort(mapApply((k, v) -> (concat(k, '__str'), JSONExtractString(v)), attributes)) AS attributes_map_str,
    mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), resource_attributes)) AS resource_attributes,
    toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id,
    observed_timestamp + toIntervalDay(if((retention_days IS NOT NULL) AND (retention_days > 0), retention_days, toInt32OrDefault(_headers.value[indexOf(_headers.name, 'retention-days')], toInt32(15)))) AS original_expiry_timestamp,
    _partition,
    _topic,
    _offset,
    toInt64OrDefault(_headers.value[indexOf(_headers.name, 'record_count')], toInt64(1)) AS _record_count,
    toInt64OrNull(_headers.value[indexOf(_headers.name, 'bytes_uncompressed')]) / _record_count AS _bytes_uncompressed,
    toInt64OrNull(_headers.value[indexOf(_headers.name, 'bytes_compressed')]) / _record_count AS _bytes_compressed,
    ifNull(pattern, '') AS pattern,
    toUInt8(ifNull(pattern_version, 0)) AS pattern_version
FROM posthog_test.kafka_logs_avro

CREATE MATERIALIZED VIEW IF NOT EXISTS posthog_test.kafka_logs_avro_kafka_metrics_mv TO posthog_test.logs_kafka_metrics
(
    `_partition` UInt32,
    `_topic` String,
    `max_offset` SimpleAggregateFunction(max, UInt64),
    `max_observed_timestamp` SimpleAggregateFunction(max, DateTime64(6)),
    `max_timestamp` SimpleAggregateFunction(max, DateTime64(6)),
    `max_created_at` SimpleAggregateFunction(max, DateTime),
    `max_lag` SimpleAggregateFunction(max, Decimal(18, 6))
)
AS SELECT
    _partition,
    _topic,
    maxSimpleState(_offset) AS max_offset,
    maxSimpleState(observed_timestamp) AS max_observed_timestamp,
    maxSimpleState(timestamp) AS max_timestamp,
    maxSimpleState(now()) AS max_created_at,
    maxSimpleState(now() - observed_timestamp) AS max_lag
FROM posthog_test.logs34
GROUP BY _partition, _topic

CREATE MATERIALIZED VIEW IF NOT EXISTS posthog_test.kafka_logs_avro_billing_metrics_mv TO posthog_test.logs_billing_metrics
(
    `team_id` Int32,
    `time_bucket` DateTime,
    `service_name` LowCardinality(String),
    `bytes_uncompressed` SimpleAggregateFunction(sum, Float64),
    `bytes_compressed` SimpleAggregateFunction(sum, Float64),
    `record_count` SimpleAggregateFunction(sum, UInt64)
)
AS SELECT
    team_id,
    time_bucket,
    service_name,
    sumSimpleState(_bytes_uncompressed) AS bytes_uncompressed,
    sumSimpleState(_bytes_compressed) AS bytes_compressed,
    sumSimpleState(1) AS record_count
FROM
(
    SELECT
        team_id,
        toStartOfInterval(timestamp, toIntervalMinute(1)) AS time_bucket,
        service_name AS service_name,
        _bytes_uncompressed,
        _bytes_compressed
    FROM posthog_test.logs34
)
GROUP BY team_id, time_bucket, service_name

US, EU, DEV

logs

CREATE TABLE IF NOT EXISTS posthog_test.logs34
(
    `time_bucket` DateTime MATERIALIZED toStartOfDay(timestamp),
    `original_expiry_timestamp` DateTime64(6),
    `uuid` String,
    `team_id` Int32,
    `trace_id` String,
    `span_id` String,
    `trace_flags` Int32,
    `timestamp` DateTime64(6) CODEC(DoubleDelta),
    `observed_timestamp` DateTime64(6),
    `created_at` DateTime64(6) MATERIALIZED now(),
    `body` String,
    `severity_text` LowCardinality(String),
    `severity_number` Int32,
    `service_name` LowCardinality(String),
    `resource_attributes` Map(LowCardinality(String), String),
    `resource_fingerprint` UInt64 MATERIALIZED cityHash64(resource_attributes),
    `instrumentation_scope` String,
    `event_name` String,
    `attributes_map_str` Map(LowCardinality(String), String),
    `level` String ALIAS severity_text,
    `mat_body_ipv4_matches` Array(String) ALIAS extractAll(body, '(\d\.((25[0-5]|(2[0-4]|1(0, 1)[0-9])(0, 1)[0-9])\.)(2, 2)([0-9]))'),
    `time_minute` DateTime ALIAS toStartOfMinute(timestamp),
    `attributes` Map(LowCardinality(String), String) ALIAS mapApply((k, v) -> (left(k, -5), v), attributes_map_str),
    `attributes_map_float` Map(LowCardinality(String), Float64) MATERIALIZED mapFilter((k, v) -> (v IS NOT NULL), mapApply((k, v) -> (concat(left(k, -5), '__float'), toFloat64OrNull(v)), attributes_map_str)),
    `attributes_map_datetime` Map(LowCardinality(String), DateTime64(6)) MATERIALIZED mapFilter((k, v) -> (v IS NOT NULL), mapApply((k, v) -> (concat(left(k, -5), '__datetime'), parseDateTimeBestEffortOrNull(v, 6)), attributes_map_str)),
    `_partition` UInt32,
    `_topic` String,
    `_offset` UInt64,
    `_bytes_uncompressed` UInt64,
    `_bytes_compressed` UInt64,
    `_record_count` UInt64,
    `pattern` String,
    `pattern_version` UInt8,
    INDEX idx_severity_text_set severity_text TYPE set(10) GRANULARITY 1,
    INDEX idx_attributes_str_keys mapKeys(attributes_map_str) TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_attributes_str_values mapValues(attributes_map_str) TYPE bloom_filter(0.001) GRANULARITY 1,
    INDEX idx_mat_body_ipv4_matches mat_body_ipv4_matches TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_body_ngram3 lower(body) TYPE ngrambf_v1(3, 25000, 2, 0) GRANULARITY 1,
    INDEX idx_uuid_bloom uuid TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_observed_minmax observed_timestamp TYPE minmax GRANULARITY 1,
    INDEX idx_timestamp_minmax timestamp TYPE minmax GRANULARITY 1,
    PROJECTION projection_aggregate_counts
    (
        SELECT
            team_id,
            time_bucket,
            toStartOfMinute(timestamp),
            service_name,
            severity_text,
            resource_fingerprint,
            count() AS event_count
        GROUP BY
            team_id,
            time_bucket,
            toStartOfMinute(timestamp),
            service_name,
            severity_text,
            resource_fingerprint
    )
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/noshard/posthog.logs34', '{replica}-{shard}')
PARTITION BY toDate(original_expiry_timestamp)
PRIMARY KEY (team_id, time_bucket, service_name, resource_fingerprint, severity_text, timestamp)
ORDER BY (team_id, time_bucket, service_name, resource_fingerprint, severity_text, timestamp)
TTL original_expiry_timestamp
SETTINGS
    allow_experimental_reverse_key = 1,
    index_granularity_bytes = 104857600,
    index_granularity = 8192,
    ttl_only_drop_parts = 1,
    add_minmax_index_for_numeric_columns = 1,
    map_serialization_version = 'with_buckets'

CREATE TABLE IF NOT EXISTS posthog_test.log_attributes2
(
    `team_id` Int32,
    `time_bucket` DateTime64(0),
    `service_name` LowCardinality(String),
    `resource_fingerprint` UInt64 DEFAULT 0,
    `attribute_key` LowCardinality(String),
    `attribute_value` String CODEC(ZSTD(5)),
    `attribute_count` SimpleAggregateFunction(sum, UInt64),
    `attribute_type` LowCardinality(String) DEFAULT 'log',
    `original_expiry_time_bucket` DateTime DEFAULT now(),
    INDEX idx_attribute_key attribute_key TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_attribute_value attribute_value TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_attribute_key_n3 attribute_key TYPE ngrambf_v1(3, 32768, 3, 0) GRANULARITY 1,
    INDEX idx_attribute_value_n3 attribute_value TYPE ngrambf_v1(3, 32768, 3, 0) GRANULARITY 1
)
ENGINE = ReplicatedAggregatingMergeTree('/clickhouse/tables/noshard/posthog.log_attributes2', '{replica}-{shard}')
PARTITION BY toDate(original_expiry_time_bucket)
ORDER BY (team_id, attribute_type, time_bucket, resource_fingerprint, attribute_key, attribute_value)
TTL original_expiry_time_bucket
SETTINGS
    deduplicate_merge_projection_mode = 'drop',
    index_granularity = 8192

CREATE TABLE IF NOT EXISTS posthog_test.logs_billing_metrics
(
    `team_id` Int32,
    `time_bucket` DateTime64(0),
    `service_name` LowCardinality(String),
    `bytes_uncompressed` SimpleAggregateFunction(sum, UInt64),
    `bytes_compressed` SimpleAggregateFunction(sum, UInt64),
    `record_count` SimpleAggregateFunction(sum, UInt64)
)
ENGINE = ReplicatedAggregatingMergeTree('/clickhouse/tables/noshard/posthog.logs_billing_metrics', '{replica}-{shard}')
PARTITION BY toYYYYMM(time_bucket)
ORDER BY (team_id, time_bucket, service_name)
SETTINGS
    deduplicate_merge_projection_mode = 'rebuild',
    index_granularity = 8192

CREATE TABLE IF NOT EXISTS posthog_test.logs_kafka_metrics
(
    `_partition` UInt32,
    `_topic` String,
    `max_offset` SimpleAggregateFunction(max, UInt64),
    `max_observed_timestamp` SimpleAggregateFunction(max, DateTime64(9)),
    `max_timestamp` SimpleAggregateFunction(max, DateTime64(9)),
    `max_created_at` SimpleAggregateFunction(max, DateTime64(9)),
    `max_lag` SimpleAggregateFunction(max, UInt64)
)
ENGINE = ReplicatedAggregatingMergeTree('/clickhouse/tables/noshard/posthog.logs_kafka_metrics', '{replica}-{shard}')
ORDER BY (_topic, _partition)

CREATE OR REPLACE TABLE posthog_test.logs_distributed AS posthog_test.logs34 ENGINE = Distributed('posthog_single_shard', 'posthog_test', 'logs34')

CREATE OR REPLACE TABLE posthog_test.log_attributes_distributed AS posthog_test.log_attributes2 ENGINE = Distributed('posthog_single_shard', 'posthog_test', 'log_attributes2')

CREATE TABLE IF NOT EXISTS posthog_test.logs_billing_metrics_distributed AS posthog_test.logs_billing_metrics ENGINE = Distributed('posthog_single_shard', 'posthog_test', 'logs_billing_metrics')

CREATE TABLE IF NOT EXISTS posthog_test.logs_kafka_metrics_distributed AS posthog_test.logs_kafka_metrics ENGINE = Distributed('posthog_single_shard', 'posthog_test', 'logs_kafka_metrics')

CREATE MATERIALIZED VIEW IF NOT EXISTS posthog_test.logs34_to_log_attributes TO posthog_test.log_attributes2
(
    `team_id` Int32,
    `time_bucket` DateTime64(0),
    `original_expiry_time_bucket` DateTime64(0),
    `service_name` LowCardinality(String),
    `resource_fingerprint` UInt64,
    `attribute_key` LowCardinality(String),
    `attribute_value` String,
    `attribute_type` LowCardinality(String),
    `attribute_count` SimpleAggregateFunction(sum, UInt64)
)
AS SELECT
    team_id,
    time_bucket,
    original_expiry_time_bucket,
    service_name,
    resource_fingerprint,
    attribute_key,
    attribute_value,
    attribute_type,
    attribute_count
FROM
(
    SELECT
        team_id AS team_id,
        toStartOfInterval(timestamp, toIntervalMinute(10)) AS time_bucket,
        toStartOfInterval(original_expiry_timestamp, toIntervalMinute(10)) AS original_expiry_time_bucket,
        service_name AS service_name,
        resource_fingerprint,
        mapFilter((k, v) -> ((length(k) < 256) AND (length(v) < 256)), attributes) AS attributes,
        arrayJoin(attributes) AS attribute,
        'log' AS attribute_type,
        attribute.1 AS attribute_key,
        attribute.2 AS attribute_value,
        sumSimpleState(1) AS attribute_count
    FROM posthog_test.logs34
    GROUP BY
        team_id,
        time_bucket,
        original_expiry_time_bucket,
        service_name,
        resource_fingerprint,
        attributes
)

CREATE MATERIALIZED VIEW IF NOT EXISTS posthog_test.logs34_to_resource_attributes TO posthog_test.log_attributes2
(
    `team_id` Int32,
    `time_bucket` DateTime64(0),
    `original_expiry_time_bucket` DateTime64(0),
    `service_name` LowCardinality(String),
    `resource_fingerprint` UInt64,
    `attribute_key` LowCardinality(String),
    `attribute_value` String,
    `attribute_type` LowCardinality(String),
    `attribute_count` SimpleAggregateFunction(sum, UInt64)
)
AS SELECT
    team_id,
    time_bucket,
    original_expiry_time_bucket,
    service_name,
    resource_fingerprint,
    attribute_key,
    attribute_value,
    attribute_type,
    attribute_count
FROM
(
    SELECT
        team_id AS team_id,
        toStartOfInterval(timestamp, toIntervalMinute(10)) AS time_bucket,
        toStartOfInterval(original_expiry_timestamp, toIntervalMinute(10)) AS original_expiry_time_bucket,
        service_name AS service_name,
        resource_fingerprint,
        arrayJoin(resource_attributes) AS attribute,
        'resource' AS attribute_type,
        attribute.1 AS attribute_key,
        attribute.2 AS attribute_value,
        sumSimpleState(1) AS attribute_count
    FROM posthog_test.logs34
    GROUP BY
        team_id,
        time_bucket,
        original_expiry_time_bucket,
        service_name,
        resource_fingerprint,
        resource_attributes
)

CREATE TABLE IF NOT EXISTS posthog_test.kafka_logs_avro
(
    `uuid` String,
    `trace_id` String,
    `span_id` String,
    `trace_flags` Int32,
    `timestamp` DateTime64(6),
    `observed_timestamp` DateTime64(6),
    `body` String,
    `severity_text` String,
    `severity_number` Int32,
    `service_name` String,
    `resource_attributes` Map(LowCardinality(String), String),
    `instrumentation_scope` String,
    `event_name` String,
    `attributes` Map(LowCardinality(String), String),
    `retention_days` Nullable(Int32),
    `pattern` Nullable(String),
    `pattern_version` Nullable(Int32)
)
ENGINE = Kafka(warpstream_logs, kafka_topic_list = 'clickhouse_logs', kafka_group_name = 'clickhouse-logs-avro-new', kafka_format = 'Avro')
SETTINGS
    kafka_skip_broken_messages = 100,
    kafka_num_consumers = 8,
    kafka_poll_timeout_ms = 3000,
    kafka_poll_max_batch_size = 1000,
    kafka_thread_per_consumer = 1,
    input_format_avro_allow_missing_fields = 1

CREATE MATERIALIZED VIEW IF NOT EXISTS posthog_test.kafka_logs34_avro_mv TO posthog_test.logs34
(
    `uuid` String,
    `trace_id` String,
    `span_id` String,
    `trace_flags` Int32,
    `timestamp` DateTime64(6),
    `observed_timestamp` DateTime64(6),
    `body` String,
    `severity_text` String,
    `severity_number` Int32,
    `service_name` String,
    `instrumentation_scope` String,
    `event_name` String,
    `attributes_map_str` Map(String, String),
    `resource_attributes` Map(String, String),
    `team_id` Int32,
    `original_expiry_timestamp` DateTime64(6),
    `_partition` UInt64,
    `_topic` LowCardinality(String),
    `_offset` UInt64,
    `_record_count` Int64,
    `_bytes_uncompressed` Nullable(Int64),
    `_bytes_compressed` Nullable(Int64),
    `pattern` String,
    `pattern_version` UInt8
)
AS SELECT
    uuid,
    trace_id,
    span_id,
    trace_flags,
    timestamp,
    observed_timestamp,
    body,
    severity_text,
    severity_number,
    service_name,
    instrumentation_scope,
    event_name,
    mapSort(mapApply((k, v) -> (concat(k, '__str'), JSONExtractString(v)), attributes)) AS attributes_map_str,
    mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), resource_attributes)) AS resource_attributes,
    toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id,
    observed_timestamp + toIntervalDay(if((retention_days IS NOT NULL) AND (retention_days > 0), retention_days, toInt32OrDefault(_headers.value[indexOf(_headers.name, 'retention-days')], toInt32(15)))) AS original_expiry_timestamp,
    _partition,
    _topic,
    _offset,
    toInt64OrDef

…truncated. See the full SQL in the workflow logs.

trunk-io · 2026-08-30T11:20:05Z

_{View Full Report ↗︎ ⋅ Docs}

stamphog

Small, focused fix within the Convex connector by an owning-team author, with tests covering the new behavior; no risky territory (schema, auth, billing, CI, deps) touched.

Gate mechanics and policy version

Gate		Result
prerequisites	✓	all clear
deny-list	✓	no deny categories matched
size	✓	10L, 1F substantive, 28L/2F incl. docs/generated/snapshots — within ceiling
tier	✓	T1-agent / T1b-small (28L, 2F, single-area, fix)
stamphog 2.0.0b4		`.stamphog/policy.yml` @ `0131b30` · reviewed head `0131b30`

…able

talyn-app · 2026-09-01T10:59:43Z

/trunk merge

Copilot

Pull request overview

This pull request improves the Convex warehouse source’s error classification so that a missing/deleted table (404 from Convex streaming export read endpoints) is treated as non-retryable and surfaced to users with an actionable message, avoiding repeated scheduled retries and unhelpful raw driver text.

Changes:

Mark 404 Client Error as a non-retryable Convex error with a user-facing remediation message.
Extend Convex source tests to assert the 404 is recognized as non-retryable and that a friendly message is selected.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

File	Description
products/warehouse_sources/backend/temporal/data_imports/sources/convex/source.py	Adds 404 to the Convex non-retryable error map with an actionable message.
products/warehouse_sources/backend/temporal/data_imports/sources/convex/tests/test_convex.py	Adds test coverage to lock in 404 classification and ensure a friendly message is surfaced.

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

…4 test Address Copilot review feedback: `test_missing_table_404_surfaces_actionable_message` claimed to mirror the finalizer's first-match selection but used a case-sensitive `key in error_msg` check, while production selection (`external_data_job.py`) uses `error_message_matches`, which lowercases both sides. Switch the test to use `error_message_matches` so it actually reflects production matching behavior.

A new stamphog review started for this PR — the fresh verdict replaces this approval.

stamphog

Approved.

Contained fix to a single warehouse connector's error classification, authored by a member of the owning team, with new tests covering the added behavior; no risky-territory surface (auth, billing, migrations, CI, deps) is touched, and the outdated Copilot comment concerns a pre-existing test pattern rather than the new production behavior.

Gate mechanics and policy version

Gate		Result
prerequisites	✓	all clear
deny-list	✓	no deny categories matched
size	✓	10L, 1F substantive, 34L/2F incl. docs/generated/snapshots — within ceiling
tier	✓	T1-agent / T1b-small (34L, 2F, single-area, fix)
stamphog 2.0.0b4		`.stamphog/policy.yml` @ `216112f` · reviewed head `216112f`

talyn-app · 2026-09-01T11:31:29Z

/trunk merge

deployment-status-posthog · 2026-09-01T12:31:09Z

Deploy status

Environment	Status	Deployed At	Workflow
dev	✅ Deployed	2026-09-01 12:31 UTC	Run
prod-us	✅ Deployed	2026-09-01 12:51 UTC	Run
prod-eu	✅ Deployed	2026-09-01 12:51 UTC	Run

pr-assigner-resolver-posthog Bot assigned Gilbert09 Aug 30, 2026

pr-assigner-resolver-posthog Bot requested a review from a team August 30, 2026 11:02

Gilbert09 added the stamphog Request AI approval (no full review) label Aug 30, 2026 — with PostHog

stamphog Bot previously approved these changes Aug 30, 2026

View reviewed changes

Merge branch 'master' into posthog/convex-missing-table-404-non-retry…

5c1ec6a

…able

Copilot AI lite review requested due to automatic review settings September 1, 2026 10:59

Copilot started reviewing on behalf of Gilbert09 September 1, 2026 11:00 View session

Copilot AI reviewed Sep 1, 2026

View reviewed changes

Comment thread products/warehouse_sources/backend/temporal/data_imports/sources/convex/tests/test_convex.py Outdated

stamphog Bot approved these changes Sep 1, 2026

View reviewed changes

trunk-io Bot merged commit 2fe2491 into master Sep 1, 2026
207 checks passed

trunk-io Bot deleted the posthog/convex-missing-table-404-non-retryable branch September 1, 2026 11:59

Navigation Menu

Uh oh!

fix(convex): stop retrying a deleted table and explain it - #91365

fix(convex): stop retrying a deleted table and explain it#91365
trunk-io[bot] merged 3 commits into
masterfrom
posthog/convex-missing-table-404-non-retryable

Gilbert09 commented Aug 30, 2026

Uh oh!

trunk-io Bot commented Aug 30, 2026 •

edited

Loading

Uh oh!

github-actions Bot commented Aug 30, 2026

Uh oh!

github-actions Bot commented Aug 30, 2026 •

edited

Loading

ClickHouse migration SQL per cloud environment

Uh oh!

trunk-io Bot commented Aug 30, 2026 •

edited

Loading

Uh oh!

stamphog Bot left a comment

Uh oh!

talyn-app Bot commented Sep 1, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

stamphog Bot left a comment

Uh oh!

talyn-app Bot commented Sep 1, 2026

Uh oh!

Uh oh!

deployment-status-posthog Bot commented Sep 1, 2026 •

edited

Loading

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

Gilbert09 commented Aug 30, 2026

Problem

Changes

How did you test this code?

Automatic notifications

Docs update

🤖 Agent context

Uh oh!

trunk-io Bot commented Aug 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

github-actions Bot commented Aug 30, 2026

Uh oh!

github-actions Bot commented Aug 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

🤖 CI report

ClickHouse migration SQL per cloud environment

Uh oh!

trunk-io Bot commented Aug 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

stamphog Bot left a comment

Choose a reason for hiding this comment

Uh oh!

talyn-app Bot commented Sep 1, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

stamphog Bot left a comment

Choose a reason for hiding this comment

Uh oh!

talyn-app Bot commented Sep 1, 2026

Uh oh!

Uh oh!

deployment-status-posthog Bot commented Sep 1, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Deploy status

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

trunk-io Bot commented Aug 30, 2026 •

edited

Loading

github-actions Bot commented Aug 30, 2026 •

edited

Loading

trunk-io Bot commented Aug 30, 2026 •

edited

Loading

deployment-status-posthog Bot commented Sep 1, 2026 •

edited

Loading